package kafka

import (
	"context"
	"errors"
	"sync"
	"time"

	"testkjcld.bhecard.com/gitlab/kit/esim/config"
	"testkjcld.bhecard.com/gitlab/kit/esim/log"
)

type Subscriber struct {
	Brokers        []string
	Topic          string
	GroupID        string
	ReplicationNum int
	Handler        func(context.Context, *Message) error
}

type SubscribeOption func(*Subscriber)

type SubscriberEngine struct {
	lock   sync.Mutex
	Logger log.Logger
	List   []*Subscriber
	sync.WaitGroup
	gctx       context.Context
	cancelFunc context.CancelFunc
}

func NewSubscriberEngine(logger log.Logger) *SubscriberEngine {
	ctx, cancelFunc := context.WithCancel(context.Background())
	return &SubscriberEngine{
		Logger:     logger,
		gctx:       ctx,
		cancelFunc: cancelFunc,
	}
}

func WithSubscriberBrokers(brokers []string) SubscribeOption {
	return func(s *Subscriber) {
		s.Brokers = brokers
	}
}

func WithSubscriberTopic(topic string) SubscribeOption {
	return func(s *Subscriber) {
		s.Topic = topic
	}
}

func WithSubscriberGroupID(groupID string) SubscribeOption {
	return func(s *Subscriber) {
		s.GroupID = groupID
	}
}

func WithSubscriberReplicationNum(n int) SubscribeOption {
	return func(s *Subscriber) {
		s.ReplicationNum = n
	}
}

func WithSubscriberHandler(f func(ctx context.Context, msg *Message) error) SubscribeOption {
	return func(s *Subscriber) {
		s.Handler = f
	}
}

func (sub *Subscriber) Check() error {
	if len(sub.Brokers) == 0 {
		return errors.New("未输入kafka brokers")
	}
	if len(sub.Topic) == 0 {
		return errors.New("未输入kafka topic")
	}
	if len(sub.GroupID) == 0 {
		return errors.New("未输入kafka groupID")
	}
	if sub.Handler == nil {
		return errors.New("未配置kafka 消息handler")
	}
	if sub.ReplicationNum == 0 {
		sub.ReplicationNum = 1
	}
	return nil
}

func (se *SubscriberEngine) Subscribe(options ...SubscribeOption) error {
	sub := new(Subscriber)
	for _, opt := range options {
		opt(sub)
	}
	if err := sub.Check(); err != nil {
		se.Logger.Errorf("订阅kafka 错误: err=%v", err)
		return err
	}
	se.lock.Lock()
	defer se.lock.Unlock()
	se.List = append(se.List, sub)
	return nil
}

func (se *SubscriberEngine) Start() {
	for _, sub := range se.List {
		for i := 0; i < sub.ReplicationNum; i++ {
			se.Add(1)
			go func(_sub *Subscriber) {
				// 启动 consumer
				defer se.Done()
				memConfig := config.NewMemConfig()
				memConfig.Set("kafka_brokers", _sub.Brokers)
				memConfig.Set("kafka_topic", _sub.Topic)
				reader := NewReader(
					WithReaderConf(memConfig),
					WithReaderLogger(se.Logger),
					WithReaderGroupID(_sub.GroupID),
					WithReaderTopic(_sub.Topic),
				)

				for {
					ctx2, cancelFunc := context.WithTimeout(se.gctx, time.Second*10)
					msg, err := reader.FetchMessage(ctx2)
					if err != nil {
						cancelFunc()
						if errors.Is(err, context.Canceled) {
							break
						}
						if !errors.Is(err, context.DeadlineExceeded) {
							reader.logger.Errorf("获取kafka消息错误: err=%v", err)
						}
						continue
					}
					cancelFunc()
					hctx := context.Background()
					// WARNING: 错误仅记录, 不做重试相关, 自行在逻辑中将消息发送到retry topic 当中
					if err := _sub.Handler(hctx, msg); err != nil {
						reader.logger.Errorf("处理kafka消息错误: err=%v", err)
					}
					if err := reader.CommitMessage(hctx, msg); err != nil {
						reader.logger.Errorc(hctx, "kafka reader commit err: err=%v", err)
					}
				}
			}(sub)
		}
	}
}

func (se *SubscriberEngine) Stop() {
	// 关闭全部消费者
	se.cancelFunc()
	se.Wait()
	se.Logger.Infof("kafka subscriber engine 已经成功关闭")
}
